package transform;

import bean.SensorReading;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;


/**
 * Rich Functions
 * 1. open()方法是 rich function 的初始化方法，当一个算子例如 map 或者 filter被调用之前 open()会被调用。
 * 2. close()方法是生命周期中的最后一个调用的方法，做一些清理工作。
 * 3. getRuntimeContext()方法提供了函数的 RuntimeContext 的一些信息，例如函数执行的并行度，任务的名字，以及 state 状态
 *
 */
public class TransDemo05_RichFunction {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<String> inputDS = env.readTextFile("D:\\CS\\MyDemo\\gmall-flink\\flink-demo\\src\\main\\resources\\sensor.txt");

        SingleOutputStreamOperator<SensorReading> dataStream = inputDS.map(line -> {
            String[] fields = line.split(",");
            return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
        });



        SingleOutputStreamOperator<Tuple2<String, Integer>> resultStream = dataStream
                .map(new RichMapFunction<SensorReading, Tuple2<String, Integer>>() {

            @Override
            public void open(Configuration parameters) throws Exception {
                // 初始化工作，一般是定义状态，或者建立数据库连接
                System.out.println("open");
            }

            @Override
            public Tuple2<String, Integer> map(SensorReading value) throws Exception {
                return new Tuple2<>(value.getId(), getRuntimeContext().getIndexOfThisSubtask());
            }

            @Override
            public void close() throws Exception {
                // 一般是关闭连接和清空状态的收尾操作
                System.out.println("close");

            }
        });

        resultStream.print();
        env.execute();

    }
}
